package com.em.meteorology.messagecore.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.jms.JmsProperties;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.*;
import java.util.HashMap;
import java.util.Map;

/**
 * 〈一句话功能简述〉<br>
 * 〈mq 消息生产者〉
 *
 * @author pupengfei
 * @create 2020/1/2
 * @since 1.0.0
 */
@Slf4j
@Service
public class MqProducer {

    @Autowired
    private JmsMessagingTemplate jmsTemplate;

    private Map<String, Destination> queuqMap = new HashMap<>();

    public static void main(String[] args) {

    }
    /**
     * 发送消息，queueName是发送到的队列 名称，message是待发送的消息
     *
     * @param queueName 队列名称
     * @param message   消息
     */
    public void sendMessage(String queueName, String message) {
        Destination destination = queuqMap.get(queueName);
        if (destination == null) {
            destination = new ActiveMQQueue(queueName);
            queuqMap.put(queueName, destination);
        }
        jmsTemplate.convertAndSend(destination, message);
    }

    /**
     * 延时发送
     *
     * @param queueName
     * @param msg
     * @param time
     */
    public void sendMessageDelay(String queueName, String msg, Long time) {
        Destination destination = queuqMap.get(queueName);
        if (destination == null) {
            destination = new ActiveMQQueue(queueName);
            queuqMap.put(queueName, destination);
        }
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        // 获取连接工厂
        ConnectionFactory connectionFactory = jmsTemplate.getConnectionFactory();
        try {
            // 获取连接
            connection = connectionFactory.createConnection();
            connection.start();
            // 获取session，true开启事务，false关闭事务
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建一个消息队列
            producer = session.createProducer(destination);
            producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
            ObjectMessage message = session.createObjectMessage(msg);
            //设置延迟时间
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
            // 发送消息
            producer.send(message);
            log.debug("发送消息：{}", message);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (producer != null) {
                    producer.close();
                }
                if (session != null) {
                    session.close();
                }
                if (connection != null) {
                    connection.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 发送topic消息，queueName是发送到的队列 名称，message是待发送的消息
     *
     * @param queueName 队列名称
     * @param message   消息
     */
    public void sendMessageTopic(String queueName, String message) {
        Destination destination = queuqMap.get(queueName);
        if (destination == null) {
            destination = new ActiveMQTopic(queueName);
            queuqMap.put(queueName, destination);
        }

        jmsTemplate.convertAndSend(destination, message);
    }

}
